package cn.com.cfae.iras.doc.analyze.mq.consumer;

import org.apache.rocketmq.client.consumer.MQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullTaskCallback;
import org.apache.rocketmq.client.consumer.PullTaskContext;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * Created by 佟媛媛 on 18/10/23.
 */
@Component
public class AnalyzeStatusConsumerPullTaskCallback implements PullTaskCallback {

    private static Logger logger = LoggerFactory.getLogger(AnalyzeStatusConsumerPullTaskCallback.class);

    private String tag;

    private String topic;

    public AnalyzeStatusConsumerPullTaskCallback() {
    }

    public AnalyzeStatusConsumerPullTaskCallback(String topic, String tag) {
        setTopic(topic);
        setTag(tag);
    }

    @Autowired
    private MessageHandler messageHandler;

    @Override
    public void doPullTask(MessageQueue mq, PullTaskContext context) {
        MQPullConsumer consumer = context.getPullConsumer();
        try {
            // 获取从哪里拉取
            long offset = consumer.fetchConsumeOffset(mq, false);
            if (offset < 0)
                offset = 0;
            PullResult pullResult = consumer.pull(mq, getTag(), offset, 1);

            switch (pullResult.getPullStatus()) {
                case FOUND:
                    List<MessageExt> msgs = pullResult.getMsgFoundList();
                    for (MessageExt messageExt : msgs) {
                        String message = new String(messageExt.getBody(), "UTF-8");
                        String topic = messageExt.getTopic();
                        String tag = messageExt.getTags();
                        try{
                            messageHandler.handleMessage(topic, tag, message);
                        }catch (Exception e){
                            logger.error(e.getMessage(),e);
                        }
                    }
                    break;
                case NO_MATCHED_MSG:
                    break;
                case NO_NEW_MSG:
                case OFFSET_ILLEGAL:
                    break;
                default:
                    break;
            }
            consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
            // 设置再过5s后重新拉取
            context.setPullNextDelayTimeMillis(5000);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    public String getTag() {
        return tag;
    }

    public void setTag(String tag) {
        this.tag = tag;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }
}
